package hzn;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class MyProducer {
    public static void main(String[] args) {
        Properties p=new Properties();
        //设置kafka集群
        p.put("bootstrap.servers","192.168.184.100:9092");
        //应答级别
        p.setProperty("acks","all");
        p.put("retries",0);
        //设置一批数据大小
        p.put("batch.size",16*1024);
        //设置累加器，也就是缓冲区大小
        p.put("buffer.memory",32*1024*1024);
        //提交延时，达到指定时间，即使缓存不满也提交
        p.put("linger.ms",1);

        p.put("key.serializer", StringSerializer.class.getName());
        p.put("value.serializer", StringSerializer.class.getName());
        p.put("partitioner.class","hzn.MyPartition");

        Producer<String,String>pr=new KafkaProducer<String, String>(p);
        for(int i=0;i<10;i++){
            //pr.send(new ProducerRecord<String, String>("first","hello"+i));
            pr.send(new ProducerRecord<String, String>("first", i+"","callBack_" + i), (m, e) -> {
                if(e==null){//发送成功
                    System.out.println("分区："+m.partition()+"--offset:"+m.offset());
                }else{
                    System.out.println("发送失败.");
                }
            });
        }

        pr.close();

    }
}
